package structural

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.graphx.{Edge, Graph, VertexId}
import org.apache.spark.rdd.RDD

object MaskDemo {

  def main(args: Array[String]): Unit = {
    //设置运行环境
    val conf = new SparkConf().setAppName("SimpleGraphX").setMaster("local")
    val sc = new SparkContext(conf)
    sc.setLogLevel("WARN")

    //设置users顶点
    val users1: RDD[(VertexId, (String, Int))] =
      sc.parallelize(Array((1L, ("Alice", 28)), (2L, ("Bob", 27)), (3L, ("Charlie", 65)),
                           (4L, ("David", 42)), (5L, ("Ed", 55)), (6L, ("Fran", 50))))

    //设置relationships边
    val relationships1: RDD[Edge[Int]] =
      sc.parallelize(Array(Edge(2L, 1L, 7), Edge(2L, 4L, 2), Edge(3L, 2L, 4), Edge(3L, 6L, 3),
                           Edge(4L, 1L, 1),Edge(5L, 2L, 2),Edge(5L, 3L, 8),Edge(5L, 6L, 3)))

    //设置users2顶点
    val users2: RDD[(VertexId, (String, Int))] =
      sc.parallelize(Array((6L, ("Oscar", 21)), (7L, ("Tom", 25)), (3L, ("Charlie", 65)),
                           (8L, ("Willian", 18)), (5L, ("Ed", 55)), (6L, ("Silva", 60))))

    //设置relationships2边
    val relationships2: RDD[Edge[Int]] =
      sc.parallelize(Array(Edge(7L, 3L, 7), Edge(7L, 8L, 2), Edge(3L, 8L, 8), Edge(3L, 6L, 3),
                           Edge(8L, 6L, 1),Edge(5L, 7L, 2),Edge(5L, 5L, 1),Edge(5L, 6L, 3)))



    val defaultUser = ("John Doe",0)
    // Build the initial Graph

    val graph1 = Graph(users1, relationships1,defaultUser)

    graph1.edges.collect.foreach(println(_))
    graph1.vertices.collect.foreach(println(_))
    println("*******************************************************************")

    val graph2 = Graph(users2, relationships2,defaultUser)
    graph2.edges.collect.foreach(println(_))
    graph2.vertices.collect.foreach(println(_))

    println("*******************************************************************")
    println("对上述的顶点数据和边数据进行修改，再创建一张图2，使得图2中有一些点和边在图1中不存在，")
    println("然后调用图1的mask方法，传入图2作为参数，观察返回的图3的信息")

    var graph3: Graph[(String, Int), Int] = graph1.mask(graph2)
    graph3.edges.collect.foreach(println(_))
    graph3.vertices.collect.foreach(println(_))
  }



}
